构建基于Serverless架构的弹性高可用音视频处理系统

简单视频处理系统

前提条件

• 函数计算：开通函数计算服务。

• 对象存储OSS：控制台创建存储空间。

操作步骤

1. 创建服务。

   1. 登录函数计算控制台，在左侧导航栏，单击服务及函数。

   2. 在顶部菜单栏，选择地域，然后在服务列表页面，单击创建服务。

   3. 在创建服务面板，填写服务名称和描述，并按需设置配置项，然后单击确定。

      本文设置服务角色为AliyunFCDefaultRole，并在权限策略中增加AliyunOSSFullAccess。

      关于创建服务的详细配置信息，请参见创建服务。

2. 创建函数。

   1. 在函数管理页面，单击创建函数。

   2. 在创建函数页面，按需选择创建函数的方式，配置以下配置项，然后单击创建。

      • 创建函数方式：使用内置运行时创建。

      • 基本设置：配置函数的基本信息，包括函数名称和请求处理程序类型。请求处理程序类型选择处理事件请求。

      • 函数代码：函数的运行环境选择Python 3.9，代码上传方式选择使用示例代码。

      • 高级配置：由于处理视频文件比较耗时，并且要考虑视频文件的大小，所以本文选择vCPU规格为4核，内存规格为8 GB，临时硬盘大小为10 GB，执行超时时间为7200秒。

      其他配置项使用默认值，关于创建函数的详细配置信息，请参见创建函数。

3. 创建OSS触发器。

   1. 在函数详情页面，单击触发器管理页签，从版本或别名下拉列表选择要创建触发器的版本或别名，然后单击创建触发器。

   2. 在创建触发器面板，填写相关信息。然后单击确定。

      配置项	操作	本文示例
      触发器类型	选择对象存储 OSS。	对象存储 OSS
      名称	填写自定义的触发器名称。	oss-trigger
      版本或别名	默认值为LATEST，如果您需要创建其他版本或别名的触发器，需先在函数详情页的版本或别名下拉列表选择该版本。关于版本和别名的简介，请参见管理版本和管理别名。	LATEST
      Bucket 名称	选择已创建的OSS Bucket。	testbucket
      文件前缀	输入要匹配的文件名称的前缀。建议您配置文件的前缀和后缀，避免触发事件嵌套循环触发引起额外费用。此外，一个Bucket的不同触发器如果指定了相同的事件类型，则前缀和后缀不能重复。详细信息，请参见OSS触发器触发规则。 重要 文件前缀不能以/开头，否则会导致OSS触发器无法被触发。	source
      文件后缀	输入要匹配的文件名称的后缀。强烈建议您配置前缀和后缀，避免触发事件嵌套循环触发引起额外费用。另外，一个Bucket的不同触发器如果指定了相同的事件类型，则前缀和后缀不能重复。详细信息，请参见OSS触发器触发规则。	mp4
      触发事件	选择一个或多个触发事件。关于对象存储OSS的事件类型，请参见OSS事件定义。	oss:ObjectCreated:PutObject, oss:ObjectCreated:PostObject, oss:ObjectCreated:CompleteMultipartUpload
      角色名称	选择AliyunOSSEventNotificationRole。 说明 如果您第一次创建该类型的触发器，则需要在单击确定后，在弹出的对话框中选择立即授权。	AliyunOSSEventNotificationRole

4. 编写函数代码。

   1. 在函数详情页面，单击函数代码页签，在代码编辑器中编写代码。

      函数需要将MP4格式的视频文件转为FLV格式，并将FLV格式视频文件保存在OSS Bucket下的dest目录。本文以Python语言为例，示例代码如下。

      # -*- coding: utf-8 -*-
      import logging
      import oss2
      import os
      import json
      import subprocess
      import shutil
      
      logging.getLogger("oss2.api").setLevel(logging.ERROR)
      logging.getLogger("oss2.auth").setLevel(logging.ERROR)
      LOGGER = logging.getLogger()
      
      
      def get_fileNameExt(filename):
          (_, tempfilename) = os.path.split(filename)
          (shortname, extension) = os.path.splitext(tempfilename)
          return shortname, extension
      
      
      def handler(event, context):
          LOGGER.info(event)
          evt = json.loads(event)
          evt = evt["events"]
          oss_bucket_name = evt[0]["oss"]["bucket"]["name"]
          object_key = evt[0]["oss"]["object"]["key"]
          
          output_dir = "dest"
          dst_format = "flv"
          shortname, _ = get_fileNameExt(object_key)
          creds = context.credentials
          auth = oss2.StsAuth(creds.accessKeyId,
                              creds.accessKeySecret, creds.securityToken)
          oss_client = oss2.Bucket(auth, 'oss-%s-internal.aliyuncs.com' %
                                   context.region, oss_bucket_name)
      
          exist = oss_client.object_exists(object_key)
          if not exist:
              raise Exception("object {} is not exist".format(object_key))
      
          input_path = oss_client.sign_url('GET', object_key, 6 * 3600)
          # m3u8 特殊处理
          rid = context.request_id
          if dst_format == "m3u8":
              return handle_m3u8(rid, oss_client, input_path, shortname, output_dir)
          else:
              return handle_common(rid, oss_client, input_path, shortname, output_dir, dst_format)
      
      
      def handle_m3u8(request_id, oss_client, input_path, shortname, output_dir):
          ts_dir = '/tmp/ts'
          if os.path.exists(ts_dir):
              shutil.rmtree(ts_dir)
          os.mkdir(ts_dir)
          transcoded_filepath = os.path.join('/tmp', shortname + '.ts')
          split_transcoded_filepath = os.path.join(
              ts_dir, shortname + '_%03d.ts')
          cmd1 = ['ffmpeg', '-y', '-i', input_path, '-c:v',
                  'libx264', transcoded_filepath]
      
          cmd2 = ['ffmpeg', '-y', '-i', transcoded_filepath, '-c', 'copy', '-map',  '0',  '-f', 'segment',
                  '-segment_list', os.path.join(ts_dir, 'playlist.m3u8'), '-segment_time', '10', split_transcoded_filepath]
      
          try:
              subprocess.run(
                  cmd1, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
      
              subprocess.run(
                  cmd2, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
      
              for filename in os.listdir(ts_dir):
                  filepath = os.path.join(ts_dir, filename)
                  filekey = os.path.join(output_dir, shortname, filename)
                  oss_client.put_object_from_file(filekey, filepath)
                  os.remove(filepath)
                  print("Uploaded {} to {}".format(filepath, filekey))
      
          except subprocess.CalledProcessError as exc:
              # if transcode fail，trigger invoke dest-fail function
              raise Exception(request_id + " transcode failure, detail: " + str(exc))
      
          finally:
              if os.path.exists(ts_dir):
                  shutil.rmtree(ts_dir)
      
              # remove ts 文件
              if os.path.exists(transcoded_filepath):
                  os.remove(transcoded_filepath)
      
          return {}
      
      
      def handle_common(request_id, oss_client, input_path, shortname, output_dir, dst_format):
          transcoded_filepath = os.path.join('/tmp', shortname + '.' + dst_format)
          if os.path.exists(transcoded_filepath):
              os.remove(transcoded_filepath)
          cmd = ["ffmpeg", "-y", "-i", input_path, transcoded_filepath]
          try:
              subprocess.run(
                  cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
      
              oss_client.put_object_from_file(
                  os.path.join(output_dir, shortname + '.' + dst_format), transcoded_filepath)
          except subprocess.CalledProcessError as exc:
              # if transcode fail，trigger invoke dest-fail function
              raise Exception(request_id + " transcode failure, detail: " + str(exc))
          finally:
              if os.path.exists(transcoded_filepath):
                  os.remove(transcoded_filepath)
          return {}

   2. 单击部署代码。

5. 测试函数代码。

   您可以通过配置函数入口参数模拟OSS事件参数进行测试，验证代码的正确性。在实际操作过程中，当发生OSS事件时，函数会自动被触发执行。

   1. 在函数详情页面，单击函数代码页签，然后单击测试函数右侧图标，从下拉列表中，选择配置测试参数。

   2. 在配置测试参数面板，选择创建新测试事件或编辑已有测试事件页签，填写事件名称和事件内容。然后单击确定。

      本文配置的事件内容如下，关于event参数具体信息，请参见步骤二：配置函数入口参数。

      {
          "events": [
              {
                  "eventName": "oss:ObjectCreated:CompleteMultipartUpload",
                  "eventSource": "acs:oss",
                  "eventTime": "2022-08-13T06:45:43.000Z",
                  "eventVersion": "1.0",
                  "oss": {
                      "bucket": {
                          "arn": "acs:oss:cn-hangzhou:123456789:testbucket",
                          "name": "testbucket",
                          "ownerIdentity": "164901546557****"
                      },
                      "object": {
                          "deltaSize": 122539,
                          "eTag": "688A7BF4F233DC9C88A80BF985AB****",
                          "key": "source/a.mp4",
                          "size": 122539
                      },
                      "ossSchemaVersion": "1.0",
                      "ruleId": "9adac8e253828f4f7c0466d941fa3db81161****"
                  },
                  "region": "cn-hangzhou",
                  "requestParameters": {
                      "sourceIPAddress": "140.205.XX.XX"
                  },
                  "responseElements": {
                      "requestId": "58F9FF2D3DF792092E12044C"
                  },
                  "userIdentity": {
                      "principalId": "164901546557****"
                  }
              }
          ]
      }

   3. 单击测试函数，在函数代码页签查看执行结果。

视频处理工作流系统

前提条件

• 开通服务并创建Bucket

  • 函数计算：开通函数计算服务。

  • 对象存储OSS：控制台创建存储空间。

  • Serverless 工作流：开通Serverless工作流。

  • 文件存储 NAS：开通文件存储服务NAS。

  • 专有网络 VPC：开通专有网络VPC。

• 配置服务角色

  • AliyunFcDefaultRole：创建服务时需要配置该角色，同时该角色需要配置权限AliyunOSSFullAccess、AliyunFnFFullAccess、AliyunFCInvocationAccess，用于调用函数、管理工作流和管理对象存储OSS。

  • AliyunOSSEventNotificationRole：OSS默认使用此角色发送事件通知。

  • fnf-execution-default-role：创建及管理工作流所需要的角色，该角色需要配置权限AliyunFCInvocationAccess、AliyunFnFFullAccess。

• Serverless Devs工具安装及配置

  • 安装Serverless Devs工具及依赖

  • 配置Serverless Devs

操作步骤